/*
* Change Logs:
* Date           Author       Notes
* 2020-10-16     chenbin      mqtt client base on paho
*/
#include <string.h>
#include <rtthread.h>
#include "small_mqtt_client.h"

#ifdef MQTT_CLIENT_RTTHREAD_PORT_DEVICE

static rt_err_t _mqtt_rtdevice_rx_indicate(rt_device_t dev, rt_size_t size)
{
	mqtt_client_port_device_t *port_device = dev->user_data;
	
	port_device->rx_size = size;
	
	return rt_sem_release(&(port_device->rx_sem));
}

static int _mqtt_rtdevice_connect(mqtt_client_t *client)
{
	mqtt_client_port_device_t *port_device = (mqtt_client_port_device_t *)client->port;
	
	port_device->device->user_data = port_device;
		
	rt_device_set_rx_indicate(port_device->device, _mqtt_rtdevice_rx_indicate);
	
	rt_device_open(port_device->device,port_device->oflag);\
	return MQTT_OK;
}

static int _mqtt_rtdevice_disconnect(mqtt_client_t *client)
{
	mqtt_client_port_device_t *port_device = (mqtt_client_port_device_t *)client->port;
	
	rt_device_close(port_device->device);
	return MQTT_OK;
}

static int _mqtt_rtdevice_recv(mqtt_client_t *client, unsigned char *buf, int len, int timeout_ms)
{
	mqtt_client_port_device_t *port_device = (mqtt_client_port_device_t *)client->port;
	uint8_t *read_buff = buf;
	uint16_t read_len = 0;
	int rc = 0;
	int read_rc = 0;
	
	do{
		read_rc = rt_device_read(port_device->device,0, (read_buff+read_len) , (len-read_len) );
		if(read_rc > 0)
		{
			read_len += read_rc;
		}else if(read_rc == 0)
		{
			rc = rt_sem_take(&(port_device->rx_sem),timeout_ms);
			if(rc == (-RT_ETIMEOUT)) 
			{
				break; //wait timeout
			}else 
			if(rc < RT_EOK)
			{
				return rc; //wait error
			}
		}else
		{
			return read_rc; //read happen An error occurred
		}
	}while(read_len < len);
	
	return read_len; //read ok
}
/*
*返回>0数据长度
*返回=0没有接收到数据
*返回-1为一般错误,网络且正常
*返回-2为等待数据超时,网络且正常
*返回-3为网络断开,网络异常
 */
//static int _mqtt_rtdevice_recv(mqtt_client_t *client, unsigned char *buf, int len, int timeout_ms)
//{
//	mqtt_client_port_device_t *port_device = (mqtt_client_port_device_t *)client->port;
//	uint8_t *read_buff = buf;
//	uint16_t read_len = 0;
//	int rc = 0;
//	
//	do{
//		read_len += rt_device_read(port_device->device,0, (read_buff+read_len) , (len-read_len) );
//		if(read_len >= len)
//		{
//			return read_len; //read ok
//		}
//		rc = rt_sem_take(&(port_device->rx_sem),timeout_ms);
//		if(port_device->rx_size == 0)
//		{
//			return MQTT_NET_FAIL;
//		}
//		if(rc < RT_EOK)
//		{
//			return MQTT_TIMEOUT;
//		}
//	}while((rc >= RT_EOK)&&(port_device->rx_size > 0));
//	
//	return rc; //happen An error occurred
//}

static int _mqtt_rtdevice_send(mqtt_client_t *client, unsigned char *buf, int len)
{
	mqtt_client_port_device_t *port_device = (mqtt_client_port_device_t *)client->port;
	int rc = 0;
	rc = rt_device_write(port_device->device,0,buf,len);
	return rc;
}

int mqtt_port_device_init(mqtt_client_port_device_t *port,const char *device_name)
{
	*(uint32_t *)(&port->base.magic) = MQTT_PORT_DEVICE;
	port->base.net_connect = _mqtt_rtdevice_connect;
	port->base.net_disconnect = _mqtt_rtdevice_disconnect;
	port->base.net_recv = _mqtt_rtdevice_recv;
	port->base.net_send = _mqtt_rtdevice_send;
	
	port->device_name = device_name;
	port->device = rt_device_find(device_name);
	if(port->device)
	{
		rt_sem_init(&(port->rx_sem), device_name, 0, RT_IPC_FLAG_FIFO);
	}
	return 0;
}

mqtt_client_port_device_t *mqtt_port_device_create(const char *device_name)
{
	mqtt_client_port_device_t *port_device = rt_malloc_align(sizeof(mqtt_client_port_device_t),4);
	if(port_device)
	{
		rt_memset(port_device,0,sizeof(mqtt_client_port_device_t));
		mqtt_port_device_init(port_device,device_name);
		return port_device;
	}
	return NULL;
}

mqtt_client_port_device_t *mqtt_port_device_get(mqtt_client_t *client)
{
	if(client->port->net_connect == _mqtt_rtdevice_connect)
	{
		return (mqtt_client_port_device_t*)client->port;
	}
	return NULL;
}

#endif
